package com.fanqwz.kotlinmvvm.domain.usecase

import com.fanqwz.kotlinmvvm.common.util.rxjava.CompletableThreadComposer
import com.fanqwz.kotlinmvvm.common.util.rxjava.FlowableThreadComposer
import com.fanqwz.kotlinmvvm.common.util.rxjava.SingleThreadComposer
import com.fanqwz.kotlinmvvm.domain.executor.PostExecutionThread
import com.fanqwz.kotlinmvvm.domain.executor.ThreadExecutor
import io.reactivex.*
import io.reactivex.disposables.CompositeDisposable
import io.reactivex.observers.DisposableCompletableObserver
import io.reactivex.observers.DisposableSingleObserver
import io.reactivex.schedulers.Schedulers
import io.reactivex.subjects.PublishSubject
import io.reactivex.subjects.ReplaySubject
import io.reactivex.subscribers.DisposableSubscriber
import javax.inject.Inject

/**
 * Created by fanqwz on 2018/3/1.
 */
interface IUseCase<R, V> {
    fun execute(arg: V): R
}

interface Disposable {
    fun dispose()
}

class UseCaseSubject<R, V, C : IUseCase<Flowable<R>, V>> @Inject constructor(
        private val threadExecutor: ThreadExecutor,
        private val postExecutionThread: PostExecutionThread,
        private val usecase: C) : IUseCase<Observable<R>, V>, Disposable {
    private val disposables = CompositeDisposable()

    override fun execute(arg: V): Observable<R> {
        val observable = usecase.execute(arg).compose(FlowableThreadComposer(Schedulers.from(threadExecutor), postExecutionThread.scheduler))
        run {
            // 使用replaysubject，防止observable返回之前subject的onnext就已经调用，此时会丢失数据
            val sub = ReplaySubject.create<R>()
            disposables.add(observable.subscribe({
                sub.onNext(it)
            }, { t: Throwable? ->
                sub.onError(t)
            }, {
                sub.onComplete()
            }))
            return sub.hide()
        }
    }

    override fun dispose() {
        if (!disposables.isDisposed)
            disposables.dispose()
    }
}

class UseCaseFlowable<R, V> @Inject constructor(
        private val threadExecutor: ThreadExecutor,
        private val postExecutionThread: PostExecutionThread,
        private val usecase: IUseCase<Flowable<R>, V>) : IUseCase<Unit, Pair<V, DisposableSubscriber<R>>>, Disposable {
    private val disposables = CompositeDisposable()

    override fun execute(arg: Pair<V, DisposableSubscriber<R>>) {
        val observable = usecase.execute(arg.first)
                .compose(FlowableThreadComposer(Schedulers.from(threadExecutor), postExecutionThread.scheduler))
        disposables.add(observable.subscribeWith(arg.second))
    }

    override fun dispose() {
        if (!disposables.isDisposed)
            disposables.dispose()
    }
}

class UseCaseCompletable<V> @Inject constructor(
        private val threadExecutor: ThreadExecutor,
        private val postExecutionThread: PostExecutionThread,
        private val usecase: IUseCase<Completable, V>) : IUseCase<Unit, Pair<V, DisposableCompletableObserver>>, Disposable {
    private val disposables = CompositeDisposable()
    override fun execute(arg: Pair<V, DisposableCompletableObserver>) {
        val observable = usecase.execute(arg.first)
                .compose(CompletableThreadComposer(Schedulers.from(threadExecutor), postExecutionThread.scheduler))
        observable.subscribe(arg.second)
        disposables.add(arg.second)
    }

    override fun dispose() {
        if (!disposables.isDisposed)
            disposables.dispose()
    }

}

class UseCaseSingle<R, V> @Inject constructor(
        private val threadExecutor: ThreadExecutor,
        private val postExecutionThread: PostExecutionThread,
        private val usecase: IUseCase<Single<R>, V>) : IUseCase<Unit, Pair<V, DisposableSingleObserver<R>>>, Disposable {
    private val disposables = CompositeDisposable()
    override fun execute(arg: Pair<V, DisposableSingleObserver<R>>) {
        val observable = usecase.execute(arg.first)
                .compose(SingleThreadComposer(Schedulers.from(threadExecutor), postExecutionThread.scheduler))
        observable.subscribe(arg.second)
        disposables.add(arg.second)
    }

    override fun dispose() {
        if (!disposables.isDisposed)
            disposables.dispose()
    }

}